package com.wuwangfu.dim;

import com.alibaba.fastjson.JSON;
import com.wuwangfu.entity.LogBean;
import com.wuwangfu.func.GeoRichMapFunc;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.ResultSet;

/**
 * @Author jcshen
 * @Date 2023-02-27
 * @PackageName:com.wuwangfu.dim
 * @ClassName: DimLogMySQL
 * @Description:
 * @Version 1.0.0
 *
 * 需求一：关联维度数据（分类名称、地理位置信息）
 *      分类名称：查询MySQL数据库
 *      地理位置信息：使用HttpClient查询高德地图
 * 知识点
 *      1、解析json，调用filter
 *      2、查询MySQL，调用map方法，new RichMapFunction，open，close
 *      3、查询高德地图，调用map方法，new RichMapFunction，open，close
 * 测试数据
 *      {"cid": 1, "money": 600.0, "longitude":116.397128,"latitude":39.916527,"oid":"o123", }
 *      "oid":"o112", "cid": 3, "money": 200.0, "longitude":118.396128,"latitude":35.916527}
 *
 *      {"oid":"o124", "cid": 2, "money": 200.0, "longitude":117.397128,"latitude":38.916527}
 *      {"oid":"o125", "cid": 3, "money": 100.0, "longitude":118.397128,"latitude":35.916527}
 *      {"oid":"o127", "cid": 1, "money": 100.0, "longitude":116.395128,"latitude":39.916527}
 *      {"oid":"o128", "cid": 2, "money": 200.0, "longitude":117.396128,"latitude":38.916527}
 *      {"oid":"o129", "cid": 3, "money": 300.0, "longitude":115.398128,"latitude":35.916527}
 *      {"oid":"o130", "cid": 2, "money": 100.0, "longitude":116.397128,"latitude":39.916527}
 *      {"oid":"o131", "cid": 1, "money": 100.0, "longitude":117.394128,"latitude":38.916527}
 *      {"oid":"o132", "cid": 3, "money": 200.0, "longitude":118.396128,"latitude":35.916527}
 *
 * MySQL建表
 *  create table tb_category(id int,name varchar(32));
 *  insert into  tb_category values(1,'图书'),(2,'手机'),(3,'家具'),(4,'家电');
 *
 *
 */
public class DimLogMySQL {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        DataStreamSource<String> line = env.socketTextStream("localhost", 8888);
        //处理数据
        SingleOutputStreamOperator<LogBean> loged = line.map(new MapFunction<String, LogBean>() {
            @Override
            public LogBean map(String value) throws Exception {
                LogBean bean = null;
                try {
                    bean = JSON.parseObject(value, LogBean.class);
                } catch (Exception e) {
                    e.printStackTrace();
                }
                return bean;
            }
        });
        //过滤 空值
        SingleOutputStreamOperator<LogBean> filtered = loged.filter(e -> e != null);
        //filtered.print();
        //关联维度信息
        SingleOutputStreamOperator<LogBean> logWithName = filtered.map(new RichMapFunction<LogBean, LogBean>() {
            //transient 不参与序列化
            private transient Connection connection;
            private transient PreparedStatement stmt;

            @Override
            public void open(Configuration parameters) throws Exception {
                connection = DriverManager.getConnection("jdbc:mysql://node03:3306/changxin?characterEncoding=UTF-8", "root", "Changxin*8");
                stmt = connection.prepareStatement("select id,name from tb_category where id=?");
            }

            @Override
            public LogBean map(LogBean value) throws Exception {
                stmt.setInt(1, value.getCid());
                ResultSet result = stmt.executeQuery();
                String name = null;
                if (result.next()) {
                    name = result.getString(2);
                }
                result.close();
                value.setName(name);
                return value;
            }

            /**
             * 关闭连接，释放资源
             *
             * @throws Exception
             */
            @Override
            public void close() throws Exception {
                if (stmt != null) {
                    stmt.close();
                }
                if (connection != null) {
                    connection.close();
                }
            }

        });
        //logWithName.print();
        //查询高德地图，获取省市信息；4924f7ef5c86a278f5500851541cdcff
        SingleOutputStreamOperator<LogBean> locationed = logWithName.map(new GeoRichMapFunc("02154c1b8927a3dbd8c5ca0f44ccdd6b"));
        locationed.print();

        env.execute();
    }
}
